package com.bestcem.xm.common.core.config.mq;

import cn.hutool.core.date.DateField;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSON;
import com.bestcem.xm.common.core.config.delaymq.dao.MqSendLogDao;
import com.bestcem.xm.common.core.config.delaymq.entity.MqSendLogDO;
import com.bestcem.xm.common.core.config.delaymq.enums.MqSendStatusEnum;
import com.bestcem.xm.common.core.config.delaymq.util.DelayMqUtil;
import com.bestcem.xm.common.core.uitls.MsgPackFormatUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import lombok.extern.slf4j.Slf4j;
import org.bson.types.ObjectId;
import org.springframework.cloud.stream.function.StreamBridge;

import javax.annotation.Resource;
import java.util.Base64;
import java.util.Date;

@Slf4j
public abstract class XmMessageProducer {

    @Resource
    private StreamBridge bridge;

    @Resource
    private MqSendLogDao mqSendLogDao;

    /**
     * 日期在 maxDayGap 之内，则立即将消息投递入redis
     */
    private static final int MAX_DAY_GAP = 7;

    @Resource
    private DelayMqUtil delayMqUtil;


    /**
     * 发送stream 消息
     *
     * @param bindingName binding name
     * @param msg         message to transfer
     * @return is success
     * @throws JsonProcessingException
     */
    public boolean send(String bindingName, Object msg) throws JsonProcessingException {
        return bridge.send(bindingName, MsgPackFormatUtil.encode(msg));
    }


    /**
     * 发送延迟消息
     *
     * @param bindingName
     * @param msg
     * @param delayTime
     */
    public void sendDelay(String bindingName, Object msg, Integer delayTime) throws JsonProcessingException {
        Date delayDate = DateUtil.offset(com.bestcem.xm.common.core.uitls.DateUtil.getCommonDate(), DateField.MILLISECOND, delayTime).toJdkDate();
        if (log.isInfoEnabled()) {
            log.info("[message producer] sendDelay message. bindingName:{}, msg:{}, delayTime: {}", bindingName, JSON.toJSONString(msg), DateUtil.formatDateTime(delayDate));
        }
        sendDelay(bindingName, msg, delayDate);
    }

    /**
     * 发送延迟消息
     *
     * @param bindingName
     * @param msg
     * @param delayDate
     */
    public void sendDelay(String bindingName, Object msg, Date delayDate) throws JsonProcessingException {
        if (log.isInfoEnabled()) {
            log.info("[message producer] sendDelay message. bindingName:{}, msg:{}, delayDate:{}", bindingName, msg, DateUtil.formatDateTime(delayDate));
        }
        // 即时消息立即发送
        Date now = com.bestcem.xm.common.core.uitls.DateUtil.getCommonDate();
        if (delayDate.compareTo(now) <= 0) {
            send(bindingName, msg);
            return;
        }
        if (log.isInfoEnabled()) {
            log.info("[message producer] sendDelay message, now :{}, delayDate:{}， msg:{}", DateUtil.formatDateTime(now), DateUtil.formatDateTime(delayDate), JSON.toJSONString(msg));
        }

        MqSendLogDO mqSendLogDO = new MqSendLogDO();
        mqSendLogDO.setId(new ObjectId().toHexString());
        mqSendLogDO.setBindingName(bindingName);
        mqSendLogDO.setSendMsg(Base64.getEncoder().encodeToString(MsgPackFormatUtil.encode(msg)));
        mqSendLogDO.setCount(0);
        mqSendLogDO.setDelayTime(delayDate);
        mqSendLogDO.setCreateTime(now);
        mqSendLogDO.setUpdateTime(now);

        // 日期在 maxDayGap 之内，则立即将消息投递入redis 否则由 定时任务在之后加入
        if (delayDate.compareTo(DateUtil.offset(now, DateField.DAY_OF_MONTH, MAX_DAY_GAP).toJdkDate()) < 0) {
            mqSendLogDO.setStatus(MqSendStatusEnum.TO_SEND.getIndex());
        } else {
            mqSendLogDO.setStatus(MqSendStatusEnum.TO_ADD_REDIS.getIndex());
        }
        mqSendLogDao.insert(mqSendLogDO);
        delayMqUtil.addToRedis(mqSendLogDO.getId(), delayDate);
    }

}
